/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.ass.rocketmq.consumer;

import com.ass.rocketmq.constant.MQTopic;
import lombok.SneakyThrows;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.*;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Log4j2
public class PushConsumer {

    private static final  String prefix =  "dev_";
    private static final MQTopic mqTopic = MQTopic.TEST;
    public static final String DEFAULT_NAMESRVADDR = "192.168.23.104:9876;192.168.23.102:9876";
    public static final String namespace = "ly";
    public static final String accessKey = "RocketMQ";
    public static final String secretKey = "123456";


    public static void main(String[] args){
        //具实验结果来看 这几个 参数 不好使
        //consumer启动后默认从什么位置开始消费
        //1、CONSUME_FROM_LAST_OFFSET：默认策略，从该列最尾开始消费，即跳过历史消息
        //2、CONSUME_FROM_FIRST_OFFSET：从队列最开始开始消费，即历史消息（还储存在 broker 的）全部消费一遍
        //3、CONSUME_FROM_TIMESTAMP：从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
    }


    @Deprecated
    @SneakyThrows
    private static void test() {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(namespace,prefix + mqTopic.tag,
                new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)));
        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
        //设置主题 所有的tag 类似kafka所有分区下消息

        consumer.subscribe(prefix + mqTopic.topic, prefix + mqTopic.tag);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        consumer.setConsumeMessageBatchMaxSize(10);
        //wrong time format 2017_0422_221800
        //时间格式 从下面的时间点开始前消费20181109221800
//        consumer.setConsumeTimestamp("20240131101738");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt message : msgs) {

                        System.out.println("消息: " + new String(message.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

    @Deprecated
    @SneakyThrows
    //官网消息轨迹例子 不太好使
    private static void test1() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_JODIE_1",true);
        consumer.subscribe("TopicTest", "*");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//        consumer.setConsumeTimestamp("20181109221800");
        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    log.info("消息：{}",new String(msg.getBody(), StandardCharsets.UTF_8));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }

}
